|
本文整理汇总了Python中sounddevice.InputStream方法的典型用法代码示例。如果您正苦于以下问题：Python sounddevice.InputStream方法的具体用法？Python sounddevice.InputStream怎么用？Python sounddevice.InputStream使用的例子？那么恭喜您, 这里精选的方法代码示例或许可以为您提供帮助。您也可以进一步了解该方法所在类sounddevice的用法示例。
在下文中一共展示了sounddevice.InputStream方法的7个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于系统推荐出更棒的Python代码示例。
示例1: record_buffer
# 需要导入模块: import sounddevice [as 别名]
# 或者: from sounddevice import InputStream [as 别名]
def record_buffer(buffer, **kwargs):
loop = asyncio.get_event_loop()
event = asyncio.Event()
idx = 0
def callback(indata, frame_count, time_info, status):
nonlocal idx
if status:
print(status)
remainder = len(buffer) - idx
if remainder == 0:
loop.call_soon_threadsafe(event.set)
raise sd.CallbackStop
indata = indata[:remainder]
buffer[idx:idx + len(indata)] = indata
idx += len(indata)
stream = sd.InputStream(callback=callback, dtype=buffer.dtype,
channels=buffer.shape[1], **kwargs)
with stream:
await event.wait() 开发者ID:spatialaudio，项目名称:python-sounddevice，代码行数:23，代码来源:asyncio_coroutines.py
示例2: __init__
# 需要导入模块: import sounddevice [as 别名]
# 或者: from sounddevice import InputStream [as 别名]
def __init__(self,
input_device: Optional[Union[int, str]] = None,
hotword: Optional[str] = None,
hotwords: Optional[List[str]] = None,
conversation_timeout: Optional[float] = 10.0,
block_duration: float = 1.0):
"""
:param input_device: PortAudio device index or name that will be used for recording speech (default: default
system audio input device).
:param hotword: When this word is detected, the plugin will trigger a
:class:`platypush.message.event.stt.HotwordDetectedEvent` instead of a
:class:`platypush.message.event.stt.SpeechDetectedEvent` event. You can use these events for hooking other
assistants.
:param hotwords: Use a list of hotwords instead of a single one.
:param conversation_timeout: If ``hotword`` or ``hotwords`` are set and ``conversation_timeout`` is set,
the next speech detected event will trigger a :class:`platypush.message.event.stt.ConversationDetectedEvent`
instead of a :class:`platypush.message.event.stt.SpeechDetectedEvent` event. You can hook custom hooks
here to run any logic depending on the detected speech - it can emulate a kind of
"OK, Google. Turn on the lights" interaction without using an external assistant (default: 10 seconds).
:param block_duration: Duration of the acquired audio blocks (default: 1 second).
"""
super().__init__()
self.input_device = input_device
self.conversation_timeout = conversation_timeout
self.block_duration = block_duration
self.hotwords = set(hotwords or [])
if hotword:
self.hotwords = {hotword}
self._conversation_event = threading.Event()
self._input_stream: Optional[sd.InputStream] = None
self._recording_thread: Optional[threading.Thread] = None
self._detection_thread: Optional[threading.Thread] = None
self._audio_queue: Optional[queue.Queue] = None
self._current_text = '' 开发者ID:BlackLight，项目名称:platypush，代码行数:39，代码来源:__init__.py
示例3: recording_thread
# 需要导入模块: import sounddevice [as 别名]
# 或者: from sounddevice import InputStream [as 别名]
def recording_thread(self, block_duration: Optional[float] = None, block_size: Optional[int] = None,
input_device: Optional[str] = None) -> None:
"""
Recording thread. It reads raw frames from the audio device and dispatches them to ``detection_thread``.
:param block_duration: Audio blocks duration. Specify either ``block_duration`` or ``block_size``.
:param block_size: Size of the audio blocks. Specify either ``block_duration`` or ``block_size``.
:param input_device: Input device
"""
assert (block_duration or block_size) and not (block_duration and block_size), \
'Please specify either block_duration or block_size'
if not block_size:
block_size = int(self.rate * self.channels * block_duration)
self.before_recording()
self.logger.debug('Recording thread started')
device = self._get_input_device(input_device)
self._input_stream = sd.InputStream(samplerate=self.rate, device=device,
channels=self.channels, dtype='int16', latency=0,
blocksize=block_size)
self._input_stream.start()
self.on_recording_started()
get_bus().post(SpeechDetectionStartedEvent())
while self._input_stream:
try:
frames = self._input_stream.read(block_size)[0]
except Exception as e:
self.logger.warning('Error while reading from the audio input: {}'.format(str(e)))
continue
self._audio_queue.put(frames)
get_bus().post(SpeechDetectionStoppedEvent())
self.on_recording_ended()
self.logger.debug('Recording thread terminated') 开发者ID:BlackLight，项目名称:platypush，代码行数:39，代码来源:__init__.py
示例4: create_stream
# 需要导入模块: import sounddevice [as 别名]
# 或者: from sounddevice import InputStream [as 别名]
def create_stream(self, device=None):
if self.stream is not None:
self.stream.close()
self.stream = sd.InputStream(
device=device, channels=1, callback=self.audio_callback)
self.stream.start() 开发者ID:spatialaudio，项目名称:python-sounddevice，代码行数:8，代码来源:rec_gui.py
示例5: inputstream_generator
# 需要导入模块: import sounddevice [as 别名]
# 或者: from sounddevice import InputStream [as 别名]
def inputstream_generator(channels=1, **kwargs):
"""Generator that yields blocks of input data as NumPy arrays."""
q_in = asyncio.Queue()
loop = asyncio.get_event_loop()
def callback(indata, frame_count, time_info, status):
loop.call_soon_threadsafe(q_in.put_nowait, (indata.copy(), status))
stream = sd.InputStream(callback=callback, channels=channels, **kwargs)
with stream:
while True:
indata, status = await q_in.get()
yield indata, status 开发者ID:spatialaudio，项目名称:python-sounddevice，代码行数:15，代码来源:asyncio_generators.py
示例6: __init__
# 需要导入模块: import sounddevice [as 别名]
# 或者: from sounddevice import InputStream [as 别名]
def __init__(self):
self.stream = sd.InputStream(
samplerate=SAMPLE_RATE,
blocksize=BLOCK_SIZE,
channels=1,
callback=self.process_frames
)
self.stream.start()
self.is_held = True
self.times_pressed = 0 开发者ID:roligheten，项目名称:AndroidMediaControlsWindows，代码行数:13，代码来源:controller.py
示例7: test_sounddevice_lib
# 需要导入模块: import sounddevice [as 别名]
# 或者: from sounddevice import InputStream [as 别名]
def test_sounddevice_lib():
import time
import numpy as np
from sounddevice import InputStream, OutputStream, sleep as sd_sleep
"""
if no portaudio installed:
Traceback (most recent call last):
File "TestSoundCard.py", line 42, in
test_sounddevice_lib()
File "TestSoundCard.py", line 5, in test_sounddevice_lib
import sounddevice as sd
File "/usr/lib/python3.6/site-packages/sounddevice.py", line 64, in
raise OSError('PortAudio library not found')
OSError: PortAudio library not found
"""
duration = 2.5 # seconds
rx_buffer = np.ones((10 ** 6, 2), dtype=np.float32)
global current_rx, current_tx
current_rx = 0
current_tx = 0
def rx_callback(indata: np.ndarray, frames: int, time, status):
global current_rx
if status:
print(status)
rx_buffer[current_rx:current_rx + frames] = indata
current_rx += frames
def tx_callback(outdata: np.ndarray, frames: int, time, status):
global current_tx
if status:
print(status)
outdata[:] = rx_buffer[current_tx:current_tx + frames]
current_tx += frames
with InputStream(channels=2, callback=rx_callback):
sd_sleep(int(duration * 1000))
print("Current rx", current_rx)
with OutputStream(channels=2, callback=tx_callback):
sd_sleep(int(duration * 1000))
print("Current tx", current_tx) 开发者ID:jopohl，项目名称:urh，代码行数:52，代码来源:TestSoundCard.py
注：本文中的sounddevice.InputStream方法示例由纯净天空整理自Github/MSDocs等开源代码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。
|